// Copyright 2016 Proyectos y Sistemas de Mantenimiento SL (eProsima).
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// ------------------------------------------------------------------
// Modification history:
// feature: Unified Transport Abstraction Layer Unified Transport Abstraction Layer
// feature: support log suppression by key and period
// feature: interfaceWhitelist supports configuring network segments
// feature: In the case of white list configuration, only the same network segment is supported to send packets
// ------------------------------------------------------------------

#include <utility>
#include <cstring>
#include <algorithm>
#include <chrono>

#include <elog/Log.hpp>
#include <deps/common/IPLocator.h>

#include <transport/TransportInterface.h>
#include <transport/TransportFactory.hpp>
#include "UDPSenderResource.hpp"
#include "UDPTransportInterface.h"
#ifdef EDDS_XDP_TRANSPORT_COMPILE
#include <lixdpsocket.hpp>
#endif

#if !defined(_WIN32)

#include <StaticTracepoint.h>
#include <tracepoint/TracePointManager.hpp>

#define SEC(name) __attribute__((section(name), used))
unsigned short trans_trans_interface_semaphore SEC(".probes");

void usdt_in_udp_transport(unsigned char* send_buffer, uint32_t send_buffer_size, uint64_t timestamp,
                           asio::ip::address& local_address, const Locator_t& remote_locator, uint32_t mask_len) {
    auto& trace_point_info = vbsutil::tracepoint::TracePointManager::get_trace_point_info();
    if (trace_point_info.trace_point_status == vbsutil::tracepoint::ON_EXPORT) {
        if (trans_trans_interface_semaphore) {
            auto src_ip_bytes = local_address.to_v4().to_bytes();
            uint32_t src_ip = static_cast<uint32_t>(src_ip_bytes[0]) | (static_cast<uint32_t>(src_ip_bytes[1]) << 8) |
                              (static_cast<uint32_t>(src_ip_bytes[2]) << 16) |
                              (static_cast<uint32_t>(src_ip_bytes[3]) << 24);
            auto dst_ip_bytes = IPLocator::getIPv4(remote_locator);
            uint32_t dst_ip = static_cast<uint32_t>(dst_ip_bytes[0]) | (static_cast<uint32_t>(dst_ip_bytes[1]) << 8) |
                              (static_cast<uint32_t>(dst_ip_bytes[2]) << 16) |
                              (static_cast<uint32_t>(dst_ip_bytes[3]) << 24);

            for (auto& info : trace_point_info.data_infos) {
                uint32_t seq_num = info.seq_num;
                uint32_t offset = info.offset;
                uint32_t length = info.length;

                vbsutil::tracepoint::UsdtInfo usdt_info = {seq_num, timestamp, trace_point_info.vbs_version, trace_point_info.domain_id,
                                                           src_ip,  dst_ip, mask_len, vbsutil::tracepoint::ROLE_PUB};
                unsigned char* usdt_msg_addr = reinterpret_cast<unsigned char*>(&usdt_info);
                unsigned char* addr = reinterpret_cast<unsigned char*>(send_buffer + offset);
                FOLLY_SDT_WITH_SEMAPHORE4(trans, trans_interface, usdt_msg_addr, trace_point_info.topic_name.c_str(), addr, length);
            }
        }
        trace_point_info.clear();
    }
}
#endif

using namespace std;
using namespace asio;

namespace vbs {
namespace transport {

using IPLocator = vbsutil::xmlparser::IPLocator;
using IPFinder = vbsutil::xmlparser::IPFinder;
using PortParameters = vbsutil::xmlparser::PortParameters;

using SenderResource = vbs::transport::SenderResource;

UDPTransportDescriptor::UDPTransportDescriptor()
    : SocketTransportDescriptor(s_maximumMessageSize, s_maximumInitialPeersRange),
      m_output_udp_socket(0),
      socket_type_("udp"),
      xdp_rx_queue_num(1) {}

bool UDPTransportDescriptor::operator==(const UDPTransportDescriptor& t) const {
    return ((this->m_output_udp_socket == t.m_output_udp_socket) && (this->non_blocking_send == t.non_blocking_send) &&
            (SocketTransportDescriptor::operator==(t)));
}

TransmitorType UDPTransportDescriptor::get_type() const {
    return TransmitorType::TRANSMITOR_TYPE_UDP;
}

UDPTransportInterface::UDPTransportInterface(int32_t transport_kind)
    : TransportInterface(transport_kind),
      mSendBufferSize(0),
      mReceiveBufferSize(0),
      first_time_open_output_channel_(true)
#ifdef EDDS_XDP_TRANSPORT_COMPILE
      ,
      socket_type_(SOCKET_TYPE_UDP)
#endif
{
}

UDPTransportInterface::~UDPTransportInterface() {}

void UDPTransportInterface::clean() {
    assert(mInputSockets.size() == 0);
}

bool UDPTransportInterface::CloseInputChannel(const Locator& locator) {

    logDebug(RTPS_MSG_OUT, "CloseInputChannel" << locator);

    std::vector<UDPChannelResource*> channel_resources;
    {
        std::unique_lock<std::recursive_mutex> scopedLock(mInputMapMutex);
        if (!IsInputChannelOpen(locator)) {
            return false;
        }

        channel_resources = std::move(mInputSockets.at(IPLocator::getPhysicalPort(locator)));
        (void)mInputSockets.erase(IPLocator::getPhysicalPort(locator));
    }

    // We now disable and release the channels
    for (UDPChannelResource* channel : channel_resources) {
        channel->disable();
        try {
            channel->release();
        } catch (const asio::system_error& e) {
            logWarning(RTPS_MSG_OUT, "channel->release() catch :" << e.what());
        }
        channel->clear();
        delete channel;
    }

    return true;
}

void UDPTransportInterface::CloseInputChannelInterrupt(const Locator& locator) {
    std::vector<UDPChannelResource*> channel_resources;
    channel_resources = std::move(mInputSockets.at(IPLocator::getPhysicalPort(locator)));
    (void)mInputSockets.erase(IPLocator::getPhysicalPort(locator));

    // We now disable and release the channels
    for (UDPChannelResource* channel : channel_resources) {
        channel->disable();
        try {
            channel->release();
        } catch (const asio::system_error& e) {
            logWarning(RTPS_MSG_OUT, "channel->release() catch :" << e.what());
        }
        channel->clear();
        delete channel;
    }
}

void UDPTransportInterface::CloseOutputChannel(vbsUDPSocket& socket) {
    socket.cancel();
    socket.close();
}

#ifdef EDDS_XDP_TRANSPORT_COMPILE
void UDPTransportInterface::CloseOutputChannel(LixdpSocket* lixdp_socket) {
    if (lixdp_socket) {
        lixdp_socket->set_shutdown(true);
    }
}
#endif

bool UDPTransportInterface::DoInputLocatorsMatch(const Locator& left, const Locator& right) const {
    return IPLocator::getPhysicalPort(left) == IPLocator::getPhysicalPort(right);
}

bool UDPTransportInterface::init() {
    if ((configuration()->sendBufferSize == 0) || (configuration()->receiveBufferSize == 0)) {
        // Check system buffer sizes.
        ip::udp::socket socket(io_service_);
        socket.open(generate_protocol());

        if (configuration()->sendBufferSize == 0) {
            socket_base::send_buffer_size option;
            socket.get_option(option);
            set_send_buffer_size(static_cast<uint32_t>(option.value()));

            if (configuration()->sendBufferSize < s_minimumSocketBuffer) {
                set_send_buffer_size(s_minimumSocketBuffer);
                mSendBufferSize = s_minimumSocketBuffer;
            }
        }

        if (configuration()->receiveBufferSize == 0) {
            socket_base::receive_buffer_size option;
            socket.get_option(option);
            set_receive_buffer_size(static_cast<uint32_t>(option.value()));

            if (configuration()->receiveBufferSize < s_minimumSocketBuffer) {
                set_receive_buffer_size(s_minimumSocketBuffer);
                mReceiveBufferSize = s_minimumSocketBuffer;
            }
        }
    }

    if (configuration()->maxMessageSize > s_maximumMessageSize) {
        elogError(RTPS_MSG_OUT, RetCode_t::RETCODE_INCONSISTENT_POLICY, "maxMessageSize cannot be greater than 65500");
        return false;
    }

    if (configuration()->maxMessageSize > configuration()->sendBufferSize) {
        elogError(RTPS_MSG_OUT, RetCode_t::RETCODE_INCONSISTENT_POLICY,
                  "maxMessageSize cannot be greater than send_buffer_size");
        return false;
    }

    if (configuration()->maxMessageSize > configuration()->receiveBufferSize) {
        elogError(RTPS_MSG_OUT, RetCode_t::RETCODE_INCONSISTENT_POLICY,
                  "maxMessageSize cannot be greater than receive_buffer_size");
        return false;
    }

    // TODO(Ricardo) Create an event that update this list.
    get_ips(currentInterfaces);
    return true;
}

bool UDPTransportInterface::IsInputChannelOpen(const Locator& locator) const {
    std::unique_lock<std::recursive_mutex> scopedLock(mInputMapMutex);
    return IsLocatorSupported(locator) &&
           (mInputSockets.find(IPLocator::getPhysicalPort(locator)) != mInputSockets.end());
}

bool UDPTransportInterface::OpenAndBindInputSockets(const Locator& locator, std::shared_ptr<ReceiverResource> receiver,
                                                    bool is_multicast, uint32_t maxMsgSize) {
    std::unique_lock<std::recursive_mutex> scopedLock(mInputMapMutex);
    bool ret = false;
    logDebug(RTPS_TRANSPORT, "OpenAndBindInputSockets " << locator);

    std::vector<std::string> vInterfaces = get_binding_interfaces_list();
    for (std::string sInterface : vInterfaces) {
        UDPChannelResource* p_channel_resource;
        p_channel_resource = CreateInputChannelResource(sInterface, locator, is_multicast, maxMsgSize, receiver);
        if (nullptr != p_channel_resource) {
            mInputSockets[IPLocator::getPhysicalPort(locator)].push_back(p_channel_resource);
            ret = true;
        } else {
            if (ret) {
                CloseInputChannelInterrupt(locator);
            }
            return false;
        }
    }
    return ret;
}

UDPChannelResource* UDPTransportInterface::CreateInputChannelResource(const std::string& sInterface,
                                                                      const Locator& locator, bool is_multicast,
                                                                      uint32_t maxMsgSize,
                                                                      std::shared_ptr<ReceiverResource> receiver) {
    UDPChannelResource* p_channel_resource = nullptr;
    vbsUDPSocket unicastSocket = createUDPSocket(io_service_);
    bool is_xdp_created = false;
#ifdef EDDS_XDP_TRANSPORT_COMPILE
    if (socket_type_ == SOCKET_TYPE_XDP && !is_multicast) {
        uint16_t port = IPLocator::getPhysicalPort(locator);
        std::string ip = IPLocator::ip_to_string(locator);
        LixdpSocket* lixdp_socket = new LixdpSocketUDP(ip, port, configuration()->xdp_rx_queue_num);
        int ret = lixdp_socket->lixdp_init(lixdp::LIXDP_MODE_RECV_ONLY);
        if (ret != lixdp::LIXDP_SUCCESS) {
            delete lixdp_socket;
            lixdp_socket = nullptr;
            if (ret == lixdp::LIXDP_ERR_BIND_FAILED) {
                logDebug(RTPS_MSG_OUT, "lixdp_init failed, ip: " << ip << ", port: " << port << ", ret: " << ret << " "
                                                                 << lixdp::LixdpSocket::lixdp_strerror(ret));
                return p_channel_resource;
            }
            logError(RTPS_MSG_OUT, "lixdp_init failed, ip: " << ip << ", port: " << port << ", ret: " << ret << " "
                                                             << lixdp::LixdpSocket::lixdp_strerror(ret));
            is_xdp_created = false;
        } else {
            is_xdp_created = true;
            logInfo(RTPS_MSG_OUT, "UDPTransport XDP (input) IP/DEV:" << sInterface << ":" << port);
            p_channel_resource =
                new UDPChannelResource(this, unicastSocket, maxMsgSize, locator, sInterface, receiver, lixdp_socket);
        }
    }
#endif
    if (!is_xdp_created) {
        uint16_t port = IPLocator::getPhysicalPort(locator);
        try {
            unicastSocket = OpenAndBindInputSocket(sInterface, port, is_multicast);
            logInfo(RTPS_MSG_OUT, "UDPTransport UDP (input) IP/DEV:" << sInterface << ":" << port);
        } catch (asio::system_error const& e) {
            (void)e;
            elogInfoT(RTPS_MSG_OUT,
                      "UDPTransport " << sInterface << " Error binding at port: " << port << " with msg: " << e.what());
            return nullptr;
        }
        p_channel_resource =
            new UDPChannelResource(this, unicastSocket, maxMsgSize, locator, sInterface, std::move(receiver));
    }
    return p_channel_resource;
}

vbsUDPSocket UDPTransportInterface::OpenAndBindUnicastOutputSocket(const ip::udp::endpoint& endpoint, uint16_t& port) {
    vbsUDPSocket socket = createUDPSocket(io_service_);
    getSocketPtr(socket)->open(generate_protocol());
    if (mSendBufferSize != 0) {
        getSocketPtr(socket)->set_option(socket_base::send_buffer_size(static_cast<int32_t>(mSendBufferSize)));
    }
    getSocketPtr(socket)->set_option(ip::multicast::hops(configuration()->TTL));
    getSocketPtr(socket)->set_option(ip::unicast::hops(configuration()->TTL));
    getSocketPtr(socket)->bind(endpoint);
    getSocketPtr(socket)->non_blocking(configuration()->non_blocking_send);
    port = getSocketPtr(socket)->local_endpoint().port();
    // auto timeout = std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::minutes(10)).count();
#ifndef _WIN32
    struct timeval timeStruct;
    timeStruct.tv_sec = 0;
    timeStruct.tv_usec = std::chrono::duration_cast<std::chrono::microseconds>(std::chrono::minutes(10)).count();
    (void)setsockopt(getSocketPtr(socket)->native_handle(), SOL_SOCKET, SO_SNDTIMEO,
                     reinterpret_cast<const char*>(&timeStruct), sizeof(timeStruct));
#endif  // ifndef _WIN32
    return socket;
}

bool UDPTransportInterface::OpenOutputChannel(SendResourceList& sender_resource_list, const Locator& locator) {
    if (!IsLocatorSupported(locator)) {
        return false;
    }
    logDebug(RTPS_MSG_OUT, "OpenOutputChannel " << locator);

    std::vector<IPFinder::info_IP> locNames;
    get_unknown_network_interfaces(sender_resource_list, locNames);

    if (locNames.empty() && (!first_time_open_output_channel_)) {
        logDebug(RTPS_MSG_OUT, "no interface??? " << locNames.empty());

        rescan_interfaces_.store(false);
        return true;
    }

    try {
        uint16_t port = configuration()->m_output_udp_socket;
        // If there is no whitelist, we can simply open a generic output socket
        // and gain efficiency.
        if (is_interface_whitelist_empty()) {
            logDebug(RTPS_MSG_OUT, "OpenOutputChannel no whitelist" << locator);

            // Create sockets for outbounding multicast for the other found network interfaces.
            if (!locNames.empty()) {
                // Create other socket for outbounding rest of interfaces.
                for (auto locIt = locNames.begin(); locIt != locNames.end(); ++locIt) {
                    uint16_t real_port = 0;
                    try {
                        vbsUDPSocket multicastSocket =
                            OpenAndBindUnicastOutputSocket(generate_endpoint((*locIt).name, 0), real_port);
                        SetSocketOutboundInterface(multicastSocket, (*locIt).name);
                        // set DF is FALSE
#if defined(_WIN32)
                        BOOL df_flag = FALSE;
                        if (setsockopt(getSocketPtr(multicastSocket)->native_handle(), IPPROTO_IP, IP_DONTFRAGMENT,
                                       (const char*)&df_flag, sizeof(df_flag)) == SOCKET_ERROR) {
                            logError(RTPS_MSG_OUT,
                                     "setsockopt failed to set IP_DONTFRAGMENT with error: " << WSAGetLastError());
                        }
#else
                        int pmtu = IP_PMTUDISC_DONT;  // 设置为允许分片
                        if (setsockopt(getSocketPtr(multicastSocket)->native_handle(), IPPROTO_IP, IP_MTU_DISCOVER,
                                       &pmtu, sizeof(pmtu)) != 0) {
                            logError(RTPS_MSG_OUT,
                                     "setsockopt failed to set IP_MTU_DISCOVER with error: " << strerror(errno));
                        }
#endif
                        sender_resource_list.emplace_back(
#ifdef EDDS_XDP_TRANSPORT_COMPILE
                            static_cast<SenderResource*>(new UDPSenderResource(*this, multicastSocket, nullptr,
                                                                               (*locIt).mask_len, false, true)));
#else
                            static_cast<SenderResource*>(
                                new UDPSenderResource(*this, multicastSocket, (*locIt).mask_len, false, true)));
#endif
                        logDebug(RTPS_MSG_OUT, "UDPTransport(output) IP/DEV:" << (*locIt).name << ":" << real_port
                                                                              << " mask:" << (*locIt).mask_len
                                                                              << " mode: UDP"
                                                                              << " without whitelist.");
                    } catch (asio::system_error const& e) {
                        (void)e;
                        elogError(RTPS_MSG_OUT, RetCode_t::RETCODE_OUT_OF_RESOURCES,
                                  "UDPTransport Error binding interface " << (*locIt).name
                                                                          << " (skipping) with msg: " << e.what());
                    }
                }
            }
            if (first_time_open_output_channel_) {
                first_time_open_output_channel_ = false;
                // We add localhost output for multicast, so in case the network cable is unplugged,
                // local participants keep receiving DATA(p) announcements Also in case that no
                // network interfaces were found
                try {
                    uint16_t real_port = 0;
                    vbsUDPSocket unicastSocket =
                        OpenAndBindUnicastOutputSocket(GenerateAnyAddressEndpoint(port), real_port);
                    getSocketPtr(unicastSocket)->set_option(ip::multicast::enable_loopback(true));
                    SetSocketOutboundInterface(unicastSocket, localhost_name());
                    // set DF is FALSE
#if defined(_WIN32)
                    BOOL df_flag = FALSE;
                    if (setsockopt(getSocketPtr(unicastSocket)->native_handle(), IPPROTO_IP, IP_DONTFRAGMENT,
                                   (const char*)&df_flag, sizeof(df_flag)) == SOCKET_ERROR) {
                        logError(RTPS_MSG_OUT,
                                 "setsockopt failed to set IP_DONTFRAGMENT with error: " << WSAGetLastError());
                    }
#else
                    int pmtu = IP_PMTUDISC_DONT;  // 设置为允许分片
                    if (setsockopt(getSocketPtr(unicastSocket)->native_handle(), IPPROTO_IP, IP_MTU_DISCOVER, &pmtu,
                                   sizeof(pmtu)) != 0) {
                        logError(RTPS_MSG_OUT,
                                 "setsockopt failed to set IP_MTU_DISCOVER with error: " << strerror(errno));
                    }
#endif
                    sender_resource_list.emplace_back(
#ifdef EDDS_XDP_TRANSPORT_COMPILE
                        static_cast<SenderResource*>(
                            new UDPSenderResource(*this, unicastSocket, nullptr, 0, false, true)));
#else
                        static_cast<SenderResource*>(new UDPSenderResource(*this, unicastSocket, 0, false, true)));
#endif
                    logDebug(RTPS_MSG_OUT, "UDPTransport(output) IP/DEV:" << localhost_name() << ":" << real_port
                                                                          << " mode: UDP"
                                                                          << " without whitelist.");
                } catch (asio::system_error const& e) {
                    (void)e;
                    elogError(RTPS_MSG_OUT, RetCode_t::RETCODE_OUT_OF_RESOURCES,
                              "UDPTransport Error binding interface " << localhost_name()
                                                                      << " (skipping) with msg: " << e.what());
                }
            }
        } else {

            create_sender_resource_with_whitelist(sender_resource_list, port);
        }
    } catch (asio::system_error const& e) {
        (void)e;
        /* TODO Que hacer?
           elogError(RTPS_MSG_OUT, RetCode_t::RETCODE_ERROR,
           "UDPTransport Error binding at port: (" <<
           IPLocator::getPhysicalPort(locator) << ")"
            << " with msg: " << e.what());
           for (auto& socket : mOutputSockets)
           {
            delete socket;
           }
           mOutputSockets.clear();
         */
        return false;
    } catch (const std::out_of_range& e) {
        (void)e;
        elogError(RTPS_MSG_OUT, RetCode_t::RETCODE_OUT_OF_RESOURCES, "UDPTransport Error catch: " << e.what());
    }
    rescan_interfaces_.store(false);
    return true;
}

Locator UDPTransportInterface::RemoteToMainLocal(const Locator& remote) const {
    if (!IsLocatorSupported(remote)) {
        return false;
    }

    Locator mainLocal(remote);
    // memset(mainLocal.address, 0x00, sizeof(mainLocal.address));
    mainLocal.set_Invalid_Address();
    return mainLocal;
}

bool UDPTransportInterface::transform_remote_locator(const Locator& remote_locator, Locator& result_locator) const {
    if (IsLocatorSupported(remote_locator)) {
        result_locator = remote_locator;
        if (!is_local_locator(result_locator)) {
            // is_local_locator will return false for multicast addresses as well as
            // remote unicast ones.
            return true;
        }

        // If we get here, the locator is a local unicast address
        if (!is_locator_allowed(result_locator)) {
            return false;
        }

        //(if the whitelist is not empty and loopback is in the whitelist, use loopback )
        Locator loopbackLocator;
        fill_local_ip(loopbackLocator);
        if (!is_interface_whitelist_empty() && is_interface_allowed(IPLocator::toIPv4string(loopbackLocator))) {
            // Loopback locator
            fill_local_ip(result_locator);
        }
        return true;
    }
    return false;
}

bool UDPTransportInterface::send(unsigned char* send_buffer, uint32_t send_buffer_size, vbsUDPSocket& socket,
                                 asio::ip::address& local_address,
                                 vbsutil::xmlparser::LocatorsIterator* destination_locators_begin,
                                 vbsutil::xmlparser::LocatorsIterator* /*destination_locators_end*/,
                                 bool only_multicast_purpose, bool whitelisted, uint32_t mask_len) {
    vbsutil::xmlparser::LocatorsIterator& it = *destination_locators_begin;
#if !defined(_WIN32)
    uint64_t timestamp = std::chrono::duration_cast<std::chrono::nanoseconds>(
                             std::chrono::high_resolution_clock::now().time_since_epoch())
                             .count();
#endif
    bool ret = false;
    logDebug(RTPS_MSG_OUT, "UDPTransport: "
                               << " send " << local_address << " to dest: " << IPLocator::to_string(*it));

    if (send_buffer_size > configuration()->sendBufferSize) {
        return false;
    }
    if (mask_len && (!only_multicast_purpose) && (!local_address.is_unspecified()) && (!local_address.is_loopback())) {
        //while (it != *destination_locators_end)
        {
            if (IsLocatorSupported(*it))  // had checked before adding into writer‘s locator_selector list
            {
                logDebug(RTPS_MSG_OUT, "UDPTransport: "
                                           << " bind " << local_address << " dest: " << IPLocator::to_string(*it));
                if (IPLocator::isMulticast(*it) || IPLocator::isLocal(*it)) {
                    logDebug(RTPS_MSG_OUT, "Dst is Multi or Local, send directly.");
                    ret |= send(send_buffer, send_buffer_size, socket, *it, only_multicast_purpose, whitelisted);
                } else {
                    bool same_net = IPLocator::isSameNetworkSegment(IPLocator::ip_to_string(*it),
                                                                    local_address.to_string(), mask_len);
                    logDebug(RTPS_MSG_OUT, "Dst is not Multi and Local. try match: "
                                               << " bind: " << local_address.to_string()
                                               << " dest: " << IPLocator::to_string(*it) << " mask len: " << mask_len);
                    if (same_net) {
                        ret |= send(send_buffer, send_buffer_size, socket, *it, only_multicast_purpose, whitelisted);
                        logDebug(RTPS_MSG_OUT, "same network send: "
                                                   << " FROM " << local_address.to_string()
                                                   << " TO: " << IPLocator::to_string(*it));
                    }
                }
            }
            //++it;
        }
    } else {
        //while (it != *destination_locators_end)
        {
            if (IsLocatorSupported(*it))  // had checked before adding into writer‘s locator_selector list
            {
                if (local_address.is_loopback() && !IPLocator::isMulticast(*it) && !IPLocator::isLocal(*it)) {
                    return false;
                }
                ret |= send(send_buffer, send_buffer_size, socket, *it, only_multicast_purpose, whitelisted);
            }

            //++it;
        }
    }
#if !defined(_WIN32)
    usdt_in_udp_transport(send_buffer, send_buffer_size, timestamp, local_address, *it, mask_len);
#endif
    return ret;
}

inline bool UDPTransportInterface::send(unsigned char* send_buffer, uint32_t send_buffer_size, vbsUDPSocket& socket,
                                        const Locator& remote_locator, bool only_multicast_purpose, bool whitelisted) {
    bool success = false;
    bool is_multicast_remote_address = IPLocator::isMulticast(remote_locator);

    if ((is_multicast_remote_address == only_multicast_purpose) || whitelisted) {
        auto destinationEndpoint = generate_endpoint(remote_locator, IPLocator::getPhysicalPort(remote_locator));

        size_t bytesSent = 0;

        try {
            asio::error_code ec;
            bytesSent =
                getSocketPtr(socket)->send_to(asio::buffer(send_buffer, send_buffer_size), destinationEndpoint, 0, ec);
            if (!!ec) {
                if ((ec.value() == asio::error::would_block) || (ec.value() == asio::error::try_again)) {
                    logWarning(RTPS_MSG_OUT, "UDP send would have blocked. Packet is dropped.");
                    return true;
                }

                elogError(RTPS_MSG_OUT, RetCode_t::RETCODE_NOT_DELIVERED,
                          ec.message() << " From " << getSocketPtr(socket)->local_endpoint()
                                       << " to ip: " << destinationEndpoint.address().to_v4().to_string()
                                       << " port: " << IPLocator::getPhysicalPort(remote_locator));
                return false;
            }
        } catch (const std::exception& error) {
            elogError(RTPS_MSG_OUT, RetCode_t::RETCODE_ERROR, "Send fail. err message: " << error.what());
            return false;
        }

        (void)bytesSent;
        logDebug(RTPS_MSG_OUT, "UDPTransport: " << bytesSent << " bytes TO endpoint: " << destinationEndpoint
                                                << " FROM " << getSocketPtr(socket)->local_endpoint());
        success = true;
    }

    return success;
}

#ifdef EDDS_XDP_TRANSPORT_COMPILE
bool UDPTransportInterface::send(unsigned char* send_buffer, uint32_t send_buffer_size, LixdpSocket* lixdp_socket,
                                 vbsutil::xmlparser::LocatorsIterator* destination_locators_begin,
                                 vbsutil::xmlparser::LocatorsIterator* /*destination_locators_end*/,
                                 bool only_multicast_purpose, bool whitelisted, uint32_t mask_len) {
    vbsutil::xmlparser::LocatorsIterator& it = *destination_locators_begin;

    LixdpSocketUDP* lixdp_socket_udp = dynamic_cast<LixdpSocketUDP*>(lixdp_socket);
    Locator_t locator = lixdp_socket_udp->get_locator();

    bool ret = false;
    logDebug(RTPS_MSG_OUT, "UDPTransport: "
                               << " send " << lixdp_socket_udp->get_locator_string()
                               << " to dest: " << IPLocator::to_string(*it));

    if (only_multicast_purpose) {
        logError(RTPS_MSG_OUT, "UDPTransport: multicast is not supported for lixdp socket");
        return false;
    }

    if (IPLocator::isMulticast(*it) || IPLocator::isLocal(*it)) {
        logError(RTPS_MSG_OUT, "Dst is Multi or Local:" << IPLocator::to_string(*it) << " not supported");
        return false;
    }

    if (IPLocator::isAny(locator) || IPLocator::isLocal(locator)) {
        logError(RTPS_MSG_OUT, "Local is not Any or Local:" << IPLocator::to_string(*it) << " not supported");
        return false;
    }

    if (mask_len) {
        if (IsLocatorSupported(*it))  // had checked before adding into writer‘s locator_selector list
        {
            logDebug(RTPS_MSG_OUT, "UDPTransport: "
                                       << " bind " << lixdp_socket_udp->get_locator_string()
                                       << " dest: " << IPLocator::to_string(*it));
            bool same_net = IPLocator::isSameNetworkSegment(IPLocator::ip_to_string(*it),
                                                            lixdp_socket_udp->get_ip_string(), mask_len);
            logDebug(RTPS_MSG_OUT, "Dst is not Multi and Local. try match: "
                                       << " bind: " << lixdp_socket_udp->get_locator_string()
                                       << " dest: " << IPLocator::to_string(*it) << " mask len: " << mask_len);
            if (same_net) {
                ret |= send(send_buffer, send_buffer_size, lixdp_socket, *it, whitelisted);
                logDebug(RTPS_MSG_OUT, "same network send: "
                                           << " FROM " << lixdp_socket_udp->get_locator_string()
                                           << " TO: " << IPLocator::to_string(*it));
            }
        }
    } else {
        if (IsLocatorSupported(*it))  // had checked before adding into writer‘s locator_selector list
        {
            logDebug(RTPS_MSG_OUT, "UDPTransport: "
                                       << " send " << lixdp_socket_udp->get_locator_string()
                                       << " to dest: " << IPLocator::to_string(*it));

            ret |= send(send_buffer, send_buffer_size, lixdp_socket, *it, whitelisted);
        }
    }

    return ret;
}

inline bool UDPTransportInterface::send(unsigned char* send_buffer, uint32_t send_buffer_size,
                                        LixdpSocket* lixdp_socket, const Locator_t& remote_locator, bool whitelisted) {
    bool success = false;

    if (whitelisted) {
        auto destinationEndpoint = generate_endpoint(remote_locator, IPLocator::getPhysicalPort(remote_locator));

        size_t bytesSent = 0;
        sockaddr_in saddr;
        if (!LixdpSocketUDP::locator_to_sockaddr(remote_locator, (struct sockaddr*)&saddr)) {
            logError(RTPS_MSG_OUT, "UDPTransport XDPSOCKET: locator_to_sockaddr failed");
            return false;
        }

        bytesSent = lixdp_socket->lixdp_send(send_buffer, send_buffer_size, (struct sockaddr*)&saddr);

        if (bytesSent <= 0) {
            elogError(RTPS_MSG_OUT, RetCode_t::RETCODE_NOT_DELIVERED,
                      lixdp::LixdpSocket::lixdp_strerror(bytesSent)
                          << " ip: " << destinationEndpoint.address().to_v4().to_string()
                          << " port: " << IPLocator::getPhysicalPort(remote_locator));
            return false;
        }
        logDebug(RTPS_MSG_OUT,
                 "UDPTransport XDPSOCKET: " << bytesSent << " bytes TO endpoint: " << destinationEndpoint << " FROM "
                                            << dynamic_cast<LixdpSocketUDP*>(lixdp_socket)->get_locator_string());
        success = true;
    }

    return success;
}
#endif

bool UDPTransportInterface::fillMetatrafficMulticastLocator(Locator& locator,
                                                            uint32_t metatraffic_multicast_port) const {
    if (locator.port == 0) {
        locator.port = metatraffic_multicast_port;
    }
    return true;
}

bool UDPTransportInterface::fillMetatrafficUnicastLocator(Locator& locator, uint32_t metatraffic_unicast_port) const {
    if (locator.port == 0) {
        locator.port = metatraffic_unicast_port;
    }
    return true;
}

bool UDPTransportInterface::configureInitialPeerLocator(Locator& locator, const PortParameters& port_params,
                                                        uint32_t domainId, LocatorList& list) const {
    if (locator.port == 0) {
        for (uint32_t i = 0; i < configuration()->maxInitialPeersRange; ++i) {
            Locator auxloc(locator);
            auxloc.port = port_params.getUnicastPort(domainId, i);

            list.push_back(auxloc);
        }
    } else {
        list.push_back(locator);
    }

    return true;
}

bool UDPTransportInterface::fillUnicastLocator(Locator& locator, uint32_t well_known_port) const {
    if (locator.port == 0) {
        locator.port = well_known_port;
    }
    return true;
}

void UDPTransportInterface::get_unknown_network_interfaces(const SendResourceList& sender_resource_list,
                                                           std::vector<IPFinder::info_IP>& locNames,
                                                           bool return_loopback) {
    locNames.clear();
    if (rescan_interfaces_) {
        get_ips(locNames, return_loopback);
        for (auto& sender_resource : sender_resource_list) {
            UDPSenderResource* udp_sender_resource = UDPSenderResource::cast(*this, sender_resource.get());
            if (nullptr != udp_sender_resource) {
                for (auto it = locNames.begin(); it != locNames.end();) {
                    if (udp_sender_resource->check_ip_address(it->locator)) {
                        it = locNames.erase(it);
                    } else {
                        ++it;
                    }
                }
            }
        }
    }
}

void UDPTransportInterface::update_network_interfaces() {
    rescan_interfaces_.store(true);
}

}  // namespace transport
}  // namespace vbs
